package mapred.config;

import java.io.IOException;

import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reporter;

public abstract class AdvancedMapReduceBase<K extends Writable, V extends Writable>
        extends MapReduceBase {

    private JobConfig config;

    private OutputCollector<K, V> output;
    
    private Reporter reporter;

    @Override
    public void configure(JobConf conf) {
        super.configure(conf);
        this.config = new JobConfig(conf);

        setUp();
    }

    public JobConfig getJobConfig() {
        return this.config;
    }
    
    public String getConfig(String key) {
        return config.get(key);
    }

    public FileSystem getFileSystem() throws IOException {
        return config.getFileSystem();
    }

    @Override
    public void close() throws IOException {
        tearDown();

        super.close();
    }

    public void output(K key, V value) throws IOException {
        this.output.collect(key, value);
    }
    
    public void incrCounter(String group, String name, long value) {
        this.reporter.incrCounter(group, name, value);
    }
    
    protected void setOutput(OutputCollector<K, V> output) {
        this.output = output;
    }
    
    protected void setReporter(Reporter reporter) {
        this.reporter = reporter;
    }
    
    public abstract void setUp();

    public abstract void tearDown() throws IOException;

}
